package com.atguigu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.realtime.app.BaseAppV2;
import com.atguigu.realtime.bean.VisitorStats;
import com.atguigu.realtime.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.logging.log4j.util.Strings;

import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.HashMap;

import static com.atguigu.realtime.common.Constant.*;

/**
 * @author chenlongStart
 * @create 2021-07-02 18:43
 * @desc
 */
public class DwsVisitorStatsApp extends BaseAppV2 {
    public static void main(String[] args) {
        new DwsVisitorStatsApp().init(4001,1,"DwsVisitorStatsApp", TOPIC_DWD_PAGE_LOG,"DwsVisitorStatsApp",TOPIC_DWM_UV,TOPIC_DWM_USER_JUMP_DETAIL);
    }
    @Override
    public void run(StreamExecutionEnvironment env, HashMap<String, DataStreamSource<String>> dsMap) {
        //解析流中的数据为同一类型，然后union
        DataStream<VisitorStats> statsStream = parseAndUnion(dsMap);
        
        //分组开窗聚合
        SingleOutputStreamOperator<VisitorStats> reducedStream = aggregateAndWindow(statsStream);

        //写入clickhouse中
        sendToClickhouse(reducedStream);
    }

    // 把聚合后的数据写入到clickhouse中
    private void sendToClickhouse(SingleOutputStreamOperator<VisitorStats> reducedStream) {
        // 对jdbc sink 进一步封装, 得到一个好用的clickhouse sink
        reducedStream
                .addSink(FlinkSinkUtil.getClickhouseSink("gmall2021","visitor_stats_2021",VisitorStats.class));
    }

    // 对union后的流开窗分组聚合
    private SingleOutputStreamOperator<VisitorStats> aggregateAndWindow(DataStream<VisitorStats> statsStream) {
        return statsStream
                .keyBy(vs -> vs.getVc() + "_" + vs.getCh() + "_" + vs.getAr() + "_" + vs.getIs_new())
                .window(TumblingEventTimeWindows.of(Time.seconds(5)))
                .sideOutputLateData(new OutputTag<VisitorStats>("late") {
                })
                .reduce(new ReduceFunction<VisitorStats>() {
                            @Override
                            public VisitorStats reduce(VisitorStats v1,
                                                       VisitorStats v2) throws Exception {
                                v1.setPv_ct(v1.getPv_ct() + v2.getPv_ct());
                                v1.setUv_ct(v1.getUv_ct() + v2.getUv_ct());
                                v1.setSv_ct(v1.getSv_ct() + v2.getSv_ct());
                                v1.setUj_ct(v1.getUj_ct() + v2.getUj_ct());
                                v1.setDur_sum(v1.getDur_sum() + v2.getDur_sum());
                                return v1;
                            }
                        },
                        new ProcessWindowFunction<VisitorStats, VisitorStats, String, TimeWindow>() {
                            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                            @Override
                            public void process(String key,
                                                Context ctx,
                                                Iterable<VisitorStats> elements,
                                                Collector<VisitorStats> out) throws Exception {

                                VisitorStats vs = elements.iterator().next();
                                long start = ctx.window().getStart();
                                long end = ctx.window().getEnd();
                                vs.setStt(sdf.format(start));
                                vs.setEdt(sdf.format(end));
                                out.collect(vs);
                            }
                        });
    }

    private DataStream<VisitorStats> parseAndUnion(HashMap<String, DataStreamSource<String>> dsMap) {
        // 1. 得到pv 和 持续访问时间
        DataStreamSource<String> pageStream = dsMap.get(TOPIC_DWD_PAGE_LOG);
        SingleOutputStreamOperator<VisitorStats> pvAndDuringTimeStatsStream = pageStream
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    JSONObject common = obj.getJSONObject("common");
                    JSONObject page = obj.getJSONObject("page");
                    return new VisitorStats(
                            "", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            0L, 1L, 0L, 0L, page.getLong("during_time"),
                            obj.getLong("ts")
                    );
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((visitor, ts) -> visitor.getTs())
                );

        // 2. uv 计算
        DataStreamSource<String> uvStream = dsMap.get(TOPIC_DWM_UV);
        SingleOutputStreamOperator<VisitorStats> uvStatsStream = uvStream
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    JSONObject common = obj.getJSONObject("common");
                    return new VisitorStats(
                            "", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            1L, 0L, 0L, 0L, 0L,
                            obj.getLong("ts")
                    );
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((visitor, ts) -> visitor.getTs())
                );

        // 3. 跳出次数
        DataStreamSource<String> ujStream = dsMap.get(TOPIC_DWM_USER_JUMP_DETAIL);
        SingleOutputStreamOperator<VisitorStats> ujStatsStream = ujStream
                .map(jsonStr -> {
                    JSONObject obj = JSON.parseObject(jsonStr);
                    JSONObject common = obj.getJSONObject("common");
                    return new VisitorStats(
                            "", "",
                            common.getString("vc"),
                            common.getString("ch"),
                            common.getString("ar"),
                            common.getString("is_new"),
                            0L, 0L, 0L, 1L, 0L,
                            obj.getLong("ts")
                    );
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((visitor, ts) -> visitor.getTs())
                );

        // 4. 进入次数的计算  来源topic_dwd_page_log
        SingleOutputStreamOperator<VisitorStats> svStatsStream = pageStream
                .flatMap(new FlatMapFunction<String, VisitorStats>() {
                    @Override
                    public void flatMap(String jsonStr,
                                        Collector<VisitorStats> out) throws Exception {
                        JSONObject obj = JSON.parseObject(jsonStr);
                        JSONObject common = obj.getJSONObject("common");
                        JSONObject page = obj.getJSONObject("page");
                        String lastPageId = page.getString("last_page_id");
                        if (Strings.isEmpty(lastPageId)) {
                            VisitorStats vs = new VisitorStats(
                                    "", "",
                                    common.getString("vc"),
                                    common.getString("ch"),
                                    common.getString("ar"),
                                    common.getString("is_new"),
                                    0L, 0L, 1L, 0L, 0L,
                                    obj.getLong("ts")
                            );
                            out.collect(vs);
                        }
                    }
                })
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                                .<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                                .withTimestampAssigner((visitor, ts) -> visitor.getTs())
                );
        return pvAndDuringTimeStatsStream.union(uvStatsStream,ujStatsStream,svStatsStream);
    }
}
